package com.nl.networkflow;


import com.nl.bean.input.UserBehavior;
import java.net.URL;
import java.sql.Timestamp;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava18.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author shihb
 * @date 2019/12/20 11:30
 * 独立访客数(uv)的统计
 */
public class UniqueVisitorWhitBloom {

  public static void main(String[] args) throws Exception {
    //1.创建执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);


    //2.source
    URL resource = PageView.class.getClassLoader().getResource("UserBehavior.csv");
    DataStreamSource<String> source = env
        .readTextFile(resource.getPath());
    //3.transform算子
    SingleOutputStreamOperator<Tuple2<Timestamp, Long>> dataStream = source
        .map(s -> {
          String[] arr = s.split(",");
          long userId = Long.parseLong(arr[0].trim());
          long itemId = Long.parseLong(arr[1].trim());
          int categoryId = Integer.parseInt(arr[2].trim());
          String behavior = arr[3].trim();
          long timestamp = Long.parseLong(arr[4].trim());
          return UserBehavior.of(userId, itemId, categoryId, behavior, timestamp);
        })
        .assignTimestampsAndWatermarks(
            new BoundedOutOfOrdernessTimestampExtractor<UserBehavior>(Time.seconds(0)) {
              @Override
              public long extractTimestamp(UserBehavior element) {
                return element.getTimestamp() * 1000;
              }
            }
        )
        .filter(userBehavior -> "pv".equals(userBehavior.getBehavior()))
        // 改变样式
        .map(new MapFunction<UserBehavior, Tuple2<String, Long>>() {
          @Override
          public Tuple2<String, Long> map(UserBehavior userBehavior) throws Exception {
            return new Tuple2<>("uv", userBehavior.getUserId());
          }
        })
        // 分组其实也是所有数据全部分在同一组
        .keyBy(tuple2 -> tuple2.f0)
        .timeWindow(Time.hours(1))
        // 定义窗口触发器,让数据来一条触发一次
        .trigger(new Trigger<Tuple2<String, Long>, TimeWindow>() {
          private Long count;
          @Override
          public TriggerResult onElement(Tuple2<String, Long> element, long timestamp,
              TimeWindow window, TriggerContext ctx) throws Exception {
            // 每5000条触发一次
            if(count==5000L){
              count=0L;
              return TriggerResult.FIRE_AND_PURGE;
            }else {
              return TriggerResult.CONTINUE;
            }
          }

          @Override
          public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx)
              throws Exception {
            return TriggerResult.CONTINUE;
          }

          @Override
          public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx)
              throws Exception {
            return TriggerResult.CONTINUE;
          }

          @Override
          public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
           
          }
        })
        // 窗口处理函数,去重处理
        .process(new UvCountWithBloom());;

    dataStream.print("uv count:");
    env.execute("uv with bloom job");

  }

  /**
   * 自定义窗口处理函数
   */
  private static class UvCountWithBloom extends ProcessWindowFunction<Tuple2<String, Long>,Tuple2<Timestamp,Long>,String,TimeWindow> {
    private BloomFilter<Long> bloomFilter;

    @Override
    public void open(Configuration parameters) throws Exception {
      super.open(parameters);
      bloomFilter=BloomFilter.create(Funnels.longFunnel(),100000000);
    }

    @Override
    public void process(String key, Context context, Iterable<Tuple2<String, Long>> elements,
        Collector<Tuple2<Timestamp, Long>> out) throws Exception {
      String redisKey = String.valueOf( context.window().getEnd());
      long count = 0L;
      // 每个窗口的count也存入redis

    }
  }
}
